package com.df.ez_game.server.logic;

import com.baidu.bjf.remoting.protobuf.Codec;
import com.baidu.bjf.remoting.protobuf.ProtobufProxy;
import com.df.ez_game.EZGameCtx;
import com.df.ez_game.server.*;
import com.df.ez_game.server.msg.client.ClientReqMsg;
import com.df.ez_game.server.msg.client.ClientRespMsg;
import com.df.ez_game.server.msg.wrapper.IntWrapper;
import com.df.ez_game.server.req_metadata.ControllerMetadatas;
import com.df.ez_game.server.req_metadata.EzMethodData;
import com.df.ez_game.tools.EzConstant;
import com.df.ez_game.tools.MsgWrapperUtils;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.channel.ChannelHandlerContext;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

@Slf4j
public class LogicMsgDispatcher implements MsgDispatcher<LogicServerNode, ClientReqMsg> {

    @Getter
    List<Disruptor<LogicMsgWrapper>> workQueues = new ArrayList<Disruptor<LogicMsgWrapper>>();
    IMsgQueueSelector selector;



    public LogicMsgDispatcher(LogicSettings settings){
        Integer workQueueSzie = settings.msgQueueDefaultSize;
        if(settings.getMsgQueueSelector()!=null){
            selector = settings.getMsgQueueSelector();
        }
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        availableProcessors*=2;
        if(workQueueSzie!=null && workQueueSzie>0){
            availableProcessors = workQueueSzie;
        }
        settings.setMsgQueueDefaultSize(availableProcessors);
        for (int i = 0; i < availableProcessors; i++) {
            int index = i;
            Disruptor<LogicMsgWrapper> disruptor = new Disruptor<LogicMsgWrapper>(new MsgEventFactory(),
                    2<<16,
                    (Runnable r)->new Thread(r,"msg_handle_thread_"+index),
                    ProducerType.MULTI,
                    //TODO 不同等待策略查看文档BlockingWaitStrategy 设置过短会导致CPU持续飙高
                    new SleepingWaitStrategy(20,200)
            );
            disruptor.handleEventsWith(new MsgWorkQueueHandler());
            disruptor.start();
            workQueues.add(disruptor);
        }
    }
    private static class MsgEventFactory implements EventFactory<LogicMsgWrapper> {

        @Override
        public LogicMsgWrapper newInstance() {
            return new LogicMsgWrapper();
        }
    }


    private static class MsgWorkQueueHandler implements EventHandler<LogicMsgWrapper> {

        @Override
        public void onEvent(LogicMsgWrapper msg, long l, boolean b) throws Exception {
            ClientReqMsg reqMsg = msg.getMsg();
            ServerNode serverNode = msg.getServerNode();
            ChannelHandlerContext ctx = msg.getCtx();
            Object param = null;
            msg.setMsg(null);//移除掉被包装对象，防止内存泄露
            msg.setServerNode(null);
            msg.setCtx(null);
            msg.setReqParamObj(null);
            if(reqMsg==null || serverNode==null || ctx==null){
                return;
            }

            LogicServerNode lsn = (LogicServerNode) serverNode;
            ControllerMetadatas metadatas = lsn.getMetadatas();

            EzMethodData logicMethodData = metadatas.getLogicMethodData(reqMsg.getCmd());

            IEzGameMsgCodec msgCodec = EzGameMsgCodecManager.getCodec(((LogicServerNode) serverNode).getSettings().getMsgProtoc());
            try {

                LogicSettings settings = ((LogicServerNode) serverNode).getSettings();
                Set<IMsgFilter> filters = settings.getFilters();
                //TODO 这个对象每个请求都new一个，能创建对象池？
                EZGameCtx reqCtx = new EZGameCtx();
                EZGameCtx.setGameCtx(reqCtx);
                try {
                    reqCtx.setReqStartTime(System.currentTimeMillis());
                    reqCtx.setMethodData(logicMethodData);
                    reqCtx.setReqMsg(reqMsg);

                    Class paramType = logicMethodData.getParamType();
                    if(paramType!=null){
                        try {
                            Object decode  = null;
//                            Object decode = logicMethodData.getParamCodec().decode(reqMsg.getData());
                            if(!logicMethodData.isParamIsBaseType()){
                                decode = msgCodec.decodeBusinessMsg(reqMsg.getData() ,logicMethodData.getParamType());
                            }else {
                                decode = msgCodec.decodeBusinessMsg(reqMsg.getData() ,logicMethodData.getParamWrapperType());
                                decode = getRawReqParamObj(logicMethodData.getParamWrapperType(),decode);
                            }
                            reqCtx.setReqParam(decode);
                        }catch (Exception e){
                            e.printStackTrace();
                            //返回请求解析出错得code
                            return;
                        }
                    }
                    param = reqCtx.getReqParam();
                    reqCtx.setChannel(ctx.channel());
                    reqCtx.setServerNode(serverNode);
                    UserSession session = lsn.getSessions().get(ctx.channel().id());
                    reqCtx.setSession(session);

                    for (IMsgFilter filter : filters) {
                        filter.oneBefore(reqCtx);
                    }
                    Object resp = logicMethodData.invoke(param);
                    Class returnType = logicMethodData.getReturnType();
                    if(returnType!=null){//需要有返回值
                        ClientRespMsg clientRespMsg = new ClientRespMsg();
                        clientRespMsg.setMsgSerial(reqMsg.getMsgSerial());
                        clientRespMsg.setCode(0);
                        clientRespMsg.setCmd(reqMsg.getCmd());
                        if(resp!=null){//TODO 检查是否要给客户端发消息
                            reqCtx.setRespObj(resp);
                            reqCtx.setRespMsg(clientRespMsg);
                            if(logicMethodData.isReturnIsBaseType()){
                                resp = getRawRetunObj(logicMethodData.getReturnWrapperType(),resp);

//                                Codec codec = ProtobufProxy.create(logicMethodData.getReturnWrapperType());

//                                byte[] respbytes = codec.encode(resp);
                                byte[] respbytes = msgCodec.encodeBusinessMsg(resp,logicMethodData.getReturnWrapperType());

                                clientRespMsg.setData(respbytes);
                            }else {
//                                Codec codec = ProtobufProxy.create(returnType);

//                                byte[] respbytes = codec.encode(resp);
                                byte[] respbytes = msgCodec.encodeBusinessMsg(resp,returnType);
                                clientRespMsg.setData(respbytes);
                            }
                        }else{
                            //返回客户端处理失败的消息
//                            log.warn("请求需要返回值，但返回值为null!cmd:"+reqMsg.getCmdId());
                            throw new RuntimeException("请求需要返回值，但返回值为null!cmd:"+reqMsg.getCmd() );
                        }
                    }
                    reqCtx.setReqEndTime(System.currentTimeMillis());
                    for (IMsgFilter filter : filters) {
                        filter.onAfter(reqCtx);
                    }
                    if(reqCtx.getRespMsg()!=null && ctx.channel().isOpen() && ctx.channel().isActive()){
                        ctx.channel().writeAndFlush(reqCtx.getRespMsg());
                    }
                }catch (Exception e){
                    reqCtx.setReqEndTime(System.currentTimeMillis());
                    reqCtx.setStatus("fail");
                    Integer code = -999;
                    if(e instanceof EzGameBusinessException){
                        EzGameBusinessException egb = (EzGameBusinessException) e;
                        reqCtx.setReason(egb.getCode()+"["+e.getMessage()+"]");
                        code = egb.getCode();
                    }else {
                        reqCtx.setReason(e.getMessage());
                    }
                    if(reqCtx.getReqMsg().getNeedResp()){
                        ClientRespMsg respMsg = reqCtx.getRespMsg();
                        if(respMsg==null){
                            respMsg = new ClientRespMsg();
                            reqCtx.setRespMsg(respMsg);
                        }
                        respMsg.setMsgSerial(reqMsg.getMsgSerial());
                        respMsg.setCmd(reqMsg.getCmd());
                        respMsg.setCode(code);
                        respMsg.setData(EzConstant.ProtoConstant.emptyBytes);
                        if( ctx.channel().isOpen() && ctx.channel().isActive()){
                            ctx.channel().writeAndFlush(reqCtx.getRespMsg());
                        }
                    }
                    e.printStackTrace();
                    for (IMsgFilter filter : filters) {
                        filter.onError(reqCtx);
                    }
                }finally {
                    for (IMsgFilter filter : filters) {
                        filter.onComplete(reqCtx);
                    }
                }


            }catch (Exception e){
                e.printStackTrace();
            }finally {
                EZGameCtx.removeContext();
            }
        }

        private Object getRawReqParamObj(Class reqParamWrapperType, Object decode) {
            if(reqParamWrapperType.equals(IntWrapper.class)){
                IntWrapper wrapper = (IntWrapper) decode;
                return wrapper.getValue();
            }
            throw new RuntimeException("未支持得请求类型:"+reqParamWrapperType.getName());
        }

        private Object getRawRetunObj(Class returnWrapperType, Object resp) {

            if(returnWrapperType.equals(IntWrapper.class)){
                IntWrapper wrapper = new IntWrapper();
                wrapper.setValue((Integer) resp);
                return wrapper;
            }
            throw new RuntimeException("未支持的返回值类型："+returnWrapperType.getName());
        }
    }



    @Override
    public void dispacher(LogicServerNode serverNode, ClientReqMsg clientReqMsg, ChannelHandlerContext ctx) {
        int nextQueueIndex = getNextQueueIndex(clientReqMsg,ctx,serverNode);
        if(nextQueueIndex<0 || nextQueueIndex>= workQueues.size()){
            log.warn("消息队列索引：{}  无法获取队列，丢弃消息! ver:{},cmd:{}",nextQueueIndex,clientReqMsg.getVer(),clientReqMsg.getCmd());
            return;
        }
        Disruptor<LogicMsgWrapper> disruptor = workQueues.get(nextQueueIndex);
        //尝试获取，获取不到，则直接丢弃消息，给客户端返回繁忙
        long cursor = 0;
        LogicMsgWrapper msg = null;
        try {
            RingBuffer<LogicMsgWrapper> ringBuffer = disruptor.getRingBuffer();
            cursor = disruptor.getRingBuffer().tryNext();
            msg = disruptor.get(cursor);
            msg.setMsg(clientReqMsg);
            msg.setServerNode(serverNode);
            msg.setCtx(ctx);
            ringBuffer.publish(cursor);
        } catch (InsufficientCapacityException e) {
            //TODO 有这个异常，表示生产过快，队列满了，处理太慢，返回繁忙的消息，直接给前端返回
            if(msg!=null){//防止环对象持有消息，造成内存泄露
                msg.setMsg(null);
                msg.setServerNode(null);
                msg.setCtx(null);
            }
            MsgCommonHandle.serverIsBusy(serverNode,clientReqMsg,ctx);
            log.error("队列已满，丢弃消息！请检查是否业务方法执行时间过长！队列index:{},消息队列个数：{}",nextQueueIndex, workQueues.size());
        }catch (Exception e){
            e.printStackTrace();
            if(msg!=null){//防止环对象持有消息，造成内存泄露
                msg.setMsg(null);
                msg.setServerNode(null);
                msg.setCtx(null);
            }
        }
    }

    private int getNextQueueIndex(ClientReqMsg readMsg, ChannelHandlerContext ctx, LogicServerNode serverNode){
        //TODO 绑定了用户id则根据用户id来分配线程，没有绑定用户id呢？根据其他，这里需要交给用户来决定怎么分配队列
        if(selector==null){
            ThreadLocalRandom random = ThreadLocalRandom.current();
            int index = random.nextInt(workQueues.size());
            return index;
        }
        int select = selector.select(serverNode, readMsg, ctx, workQueues.size());
        if(select<0){
            select = 0;
            log.warn("业务线程选择器计算得索引小于0，使用索引0");
        }
        return select%workQueues.size();
    }

}
